#   todo 对数据进行处理，清洗
import pandas as pd
from hdfs.client import Client
import pymysql


#   修改pandas的内置参数，让其显示所有的行和列
#   显示所有列数据
pd.set_option('display.max_columns',None)
#   显示所有行数据
pd.set_option('display.max_rows',None)
#   设置value的长度
pd.set_option('max_colwidth',100)
#   设置1000列时才换行
pd.set_option('display.width',1000)

#   todo 1.去除重复数据
data=pd.read_csv('source.csv')
# print("数据去重前",len(data))
data.drop_duplicates(inplace=True)
# print("数据去重后",len(data))

#   todo 2.去除空数据
print("去除空值前",len(data))
data=data.dropna()
print("去除空值后",len(data))

#   todo 3.去除地点字段里面的异常值(存在大量时间)
# print(data['place'])
data=data[  data['place'].str.contains("\|") ]
print(data['place'])


#   todo 4.提取门票最高价和最低价
data['min']=data['price'].apply(lambda x: x.split("-")[0] if "-" in x else x)
data['max']=data['price'].apply(lambda x: x.split("-")[1] if "-" in x else x)
print(data[ ['min','price','max']])

#   todo 5.删除列
data=data.drop('statu',axis=1)
print(data.columns)

#   todo 6.处理艺人名字列
#   todo 首先将艺人两个字去掉
data['users']=data['users'].apply(lambda x:x.split("艺人：")[1])
#   todo 然后如果是存在多个艺人的话就将其命名为众多明星
data['users']=data['users'].apply(lambda x: "众多明星" if len(x.split("、")) != 1 else x)
print(data['users'])


data.to_csv('dwd.csv',index=False,header=True)
#   todo 下面将数据写入mysql和hdfs
#   todo 把数据写入到hdfs
#   创建一个hdfs的客户端实例(9870是web访问hdfs的端口号)
client=Client('http://192.168.40.110:9870')
#   hdfs上面存储的路径
hdfs_path="/ljy/"
#   使用客户端的upload方法将文件上传到hdfs
# client.upload(hdfs_path=hdfs_path,local_path='dwd.csv',overwrite=True)
print("处理完的数据已经保存到hdfs,保存的路径为:/ljy/dwd.csv........................")

print(data['price'])
#     将处理好的数据写入mysql
#   建立与mysql的连接
connect=pymysql.connect(host='192.168.40.110',user='root',password='123456',database='ljy',charset='utf8')
#   通过连接创建游标对象
cursor=connect.cursor()
#   定义sql如果存在表就删除表(data_processing:数据处理)
drop_table_sql='drop table if exists dwd;'
cursor.execute(drop_table_sql)
#   定义创建表格的sql
CreateTable_sql="""
create table if not exists dwd(
            id int primary key AUTO_INCREMENT,
            city varchar(500),
            title varchar(500),
            info_url varchar(1000),
            users varchar(500),
            image_url varchar(500),
            place varchar(500),
            time varchar(500),
            price varchar(500),
            min int,
            max int
);
"""
#   执行创建表的语句
cursor.execute(CreateTable_sql)
#   定义插入数据的sql语句
insert_sql="""
    insert into dwd(
         city,title,info_url,users,image_url,place,time,price,min,max)
    values(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s);
"""
#   换成list[row]的形式方便写入mysql
data_list=data.values.tolist()
#   执行插入数据
cursor.executemany(insert_sql,data_list)

#   提交任务(只有提交了任务才会执行sql)
connect.commit()
#   关闭游标对象
cursor.close()
#   关闭连接
connect.close()
print("处理完的数据已经保存到mysql的yds数据库的dwd表..............")




